package com.atguigu.flink.tableapi;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Json;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.table.descriptors.Schema;
import org.apache.kafka.clients.consumer.ConsumerConfig;

/**
 * Created by Smexy on 2023/3/4
 *
 *  读取一个文件，将文件中的数据映射为 Table
 */
public class Demo7_WriteKafka
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction());

        Table table = tableEnvironment.fromDataStream(ds);


        //准备一个连接器的描述 充当生产者
        Kafka kafka = new Kafka()
            .property(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092")
            .topic("topicC")
            //.sinkPartitionerFixed()
            //轮询地讲数据写入分区
            .sinkPartitionerRoundRobin()
            //通用版本
            .version("universal")
            ;

        //表的结构
        Schema schema = new Schema().field("id", "STRING")
                                   .field("ts", DataTypes.BIGINT())
                                    .field("vc", DataTypes.INT());

        tableEnvironment.connect(kafka)
                        //表连接的kafka中数据的格式
                        .withFormat(new Json())
                        .withSchema(schema)
                        //为表起一个临时的名字
                        .createTemporaryTable("t1");


        table.executeInsert("t1");

    }
}
